package com.itheima.rpc.client.request;

import com.itheima.rpc.cache.ServiceProviderCache;
import com.itheima.rpc.client.cluster.DefaultStartegyProvider;
import com.itheima.rpc.client.cluster.LoadBalanceStrategy;
import com.itheima.rpc.data.RpcRequest;
import com.itheima.rpc.data.RpcResponse;
import com.itheima.rpc.enums.StatusEnum;
import com.itheima.rpc.exception.RpcException;
import com.itheima.rpc.netty.codec.FrameDecoder;
import com.itheima.rpc.netty.codec.FrameEncoder;
import com.itheima.rpc.netty.codec.RpcRequestEncoder;
import com.itheima.rpc.netty.codec.RpcResponseDecoder;
import com.itheima.rpc.netty.handler.RpcResponseHandler;
import com.itheima.rpc.netty.request.ChannelMapping;
import com.itheima.rpc.netty.request.RequestPromise;
import com.itheima.rpc.netty.request.RpcRequestHolder;
import com.itheima.rpc.provider.ServiceProvider;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.concurrent.DefaultThreadFactory;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import java.util.List;
import java.util.concurrent.ExecutionException;

/**
 * @desc 发送请求管理类
 * @author:ivan.liu
 */
@Component
@Slf4j
public class RpcRequestManager {

    @Autowired
    private ServiceProviderCache cache;

    @Autowired
    private DefaultStartegyProvider defaultStartegyProvider;

    public RpcResponse sendRequest(RpcRequest rpcRequest){
        //从缓存中获取提供者信息
        List<ServiceProvider> providerList = cache.get(rpcRequest.getClassName());
        if(CollectionUtils.isEmpty(providerList)){
            log.info("没有提供{}服务的服务端",rpcRequest.getClassName());
            throw new RpcException(StatusEnum.NOT_FOUND_SERVICE_PROVINDER);
        }
        //选取一个provider
        LoadBalanceStrategy strategy = defaultStartegyProvider.getStrategy();
        ServiceProvider serviceProvider = strategy.select(providerList);
        if(serviceProvider!=null){
            return this.requestByNetty(rpcRequest,serviceProvider);
        }else {
            throw new RpcException(StatusEnum.NOT_FOUND_SERVICE_PROVINDER);
        }
    }

    private RpcResponse requestByNetty(RpcRequest rpcRequest, ServiceProvider serviceProvider) {
        try {
            Channel channel = RpcRequestHolder.getChannel(serviceProvider.getServerIp(), serviceProvider.getRpcPort());
            if(channel==null||!channel.isActive()){
                //创建group线程池
                EventLoopGroup group=new NioEventLoopGroup(0,new DefaultThreadFactory("client"));
                Bootstrap bootstrap=new Bootstrap();
                bootstrap
                        .group(group)
                        .channel(NioSocketChannel.class)
                        .handler(new ChannelInitializer<SocketChannel>() {
                            @Override
                            protected void initChannel(SocketChannel socketChannel) throws Exception {
                                ChannelPipeline pipeline = socketChannel.pipeline();
                                pipeline.addLast("frameEncoder",new FrameEncoder());
                                pipeline.addLast("requestEncoder",new RpcRequestEncoder());
                                pipeline.addLast("frameDecoder",new FrameDecoder());
                                pipeline.addLast("responseDecoder",new RpcResponseDecoder());
                                pipeline.addLast("responseHandler",new RpcResponseHandler());
                            }
                        });
                ChannelFuture future = bootstrap.connect(serviceProvider.getServerIp(), serviceProvider.getRpcPort()).sync();
                if(future.isSuccess()){
                    channel = future.channel();
                    ChannelMapping channelMapping = new ChannelMapping(serviceProvider.getServerIp(), serviceProvider.getRpcPort(), channel);
                    RpcRequestHolder.addChannelMapping(channelMapping);
                }
            }
            RequestPromise requestPromise=new RequestPromise(channel.eventLoop());
            RpcRequestHolder.addRequestPromise(rpcRequest.getRequestId(),requestPromise);
            channel.writeAndFlush(rpcRequest);
            //阻塞时等待数据
            return (RpcResponse) requestPromise.get();
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        } finally {
            RpcRequestHolder.removeRequestPromise(rpcRequest.getRequestId());
        }
        return new RpcResponse();
    }
}
